package com.comp6231.replica3.server;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.FileAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;
import org.omg.CORBA.ORB;

import com.comp6231.replica3.common.*;
import com.comp6231.replica3.common.police_station_idl.*;

public class StationServerImpl extends StationServerInterfacePOA {

	public static final int UDP_BUFFER_SIZE = 1000;

	public static final String REQUEST_RECORDS_COUNT = "record_count";

	private ConcurrentHashMap<Character, ArrayList<Record>> database;

	private static int maxCRecID = 10000 - 1;

	private static int maxMRecID = 10000 - 1;

	public String name;

	public int udpPort;

	private Logger log;

	private HashMap<String, Integer> serversInfo;

	private ORB orb;

	public StationServerImpl(String name, int udpPort, ArrayList<Record> records) {
		this.orb = ORB.init(new String[] {}, null);
		this.udpPort = udpPort;
		this.name = name;
		database = new ConcurrentHashMap<Character, ArrayList<Record>>(26);
		for (char c = 'A'; c <= 'Z'; c++) {
			database.put(c, new ArrayList<Record>());
		}
		// initDatabase(records);
		initLogger();
		readServersInfo();
		UDPServer udpServer = new UDPServer(udpPort);
		new Thread(udpServer).start();
	}

	/**
	 * read servers' name and UDP port
	 */
	private void readServersInfo() {
		serversInfo = new HashMap<String, Integer>();
		serversInfo.put("SPVM", 2001);
		serversInfo.put("SPL", 2002);
		serversInfo.put("SPB", 2003);
		serversInfo.put("spvm", 2001);
		serversInfo.put("spl", 2002);
		serversInfo.put("spb", 2003);
	}

	/**
	 * gets the server name associated with this port number
	 * 
	 * @param port
	 * @return server name
	 */
	private String getServerNameByPort(int port) {
		if (port == 2001) {
			return "spvm";
		}
		if (port == 2002) {
			return "spl";
		}
		if (port == 2003) {
			return "spb";
		}
		return "unkown";
	}

	/**
	 * read CORBA object from file
	 * 
	 * @param orb
	 * @param name
	 * @return
	 */
	private StationServerInterface readServerObjects(String name) {
		BufferedReader br;
		try {
			br = new BufferedReader(new FileReader(name + ".ior"));
			String ior = br.readLine();
			br.close();

			org.omg.CORBA.Object o = orb.string_to_object(ior);

			return StationServerInterfaceHelper.narrow(o);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * initialize database
	 * 
	 * @param records
	 */
	private void initDatabase(ArrayList<Record> records) {
		for (Record record : records) {
			if (record instanceof CriminalRecord) {
				maxCRecID++;
				record.id = String.format("CR%1$05d", maxCRecID);
			} else {
				maxMRecID++;
				record.id = String.format("MR%1$05d", maxMRecID);
			}
			ArrayList<Record> list = database.get(Character
					.toUpperCase(record.lastName.charAt(0)));
			list.add(record);
		}
	}

	/**
	 * initialize logger
	 */
	private void initLogger() {
		this.log = Logger.getLogger(this.name);
		FileAppender fa = new FileAppender();
		fa.setName(this.name);
		fa.setFile(System.getProperty("user.dir")
				+ "/src/com/comp6231/replica3/logs/" + name + ".log");
		fa.setLayout(new PatternLayout("%d %m%n"));
		fa.setThreshold(Level.ALL);
		fa.setAppend(false);
		fa.activateOptions();
		this.log.addAppender(fa);
	}

	/**
	 * get records count for this server
	 * 
	 * @return
	 */
	protected synchronized int getRecordsCount() {
		int count = 0;
		for (ArrayList<Record> list : database.values()) {
			count += list.size();
		}
		return count;
	}

	/**
	 * determines if a criminal record already exists in database
	 * 
	 * @param record
	 * @return
	 */
	private boolean duplicateCRecord(CriminalRecord record) {
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));
		synchronized (list) {
			for (Record rec : list) {
				if (rec.equals(record)) {
					return true;
				}
			}
			return false;
		}
	}

	/**
	 * determines if a missing record already exists in database
	 * 
	 * @param record
	 * @return
	 */
	private boolean duplicateMRecord(MissingRecord record) {
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));
		synchronized (list) {
			for (Record rec : list) {
				if (rec.equals(record)) {
					return true;
				}
			}
			return false;
		}
	}

	/**
	 * assigns an ID to the given criminal record
	 * 
	 * @param record
	 * @return true if an ID assigned successfully, false if id exceeded max
	 *         value
	 */
	private synchronized boolean assignCRecID(CriminalRecord record) {
		maxCRecID++;
		if (maxCRecID > 99999) {
			return false;
		} else {
			record.id = String.format("CR%1$05d", maxCRecID);
			return true;
		}
	}

	/**
	 * assigns an ID to the given missing record
	 * 
	 * @param record
	 * @return true if an ID assigned successfully, false if id exceeded max
	 *         value
	 */
	private synchronized boolean assignMRecID(MissingRecord record) {
		maxMRecID++;
		if (maxMRecID > 99999) {
			return false;
		} else {
			record.id = String.format("MR%1$05d", maxMRecID);
			return true;
		}
	}

	/**
	 * creates a criminal record
	 * 
	 * @return true if record has been created successfully, false otherwise
	 */
	@Override
	public boolean createCRecord(String firstName, String lastName,
			String description, int status, String officerID) {

		CriminalRecord record = new CriminalRecord(null, firstName, lastName,
				description, EnumHelper.convertToRecordStatus(status));

		log.info("Officer " + officerID
				+ " requested to create a criminal record.");
		log.info(record);

		if (!officerID.startsWith(name.toUpperCase())) {
			log.info("Officer " + officerID
					+ " is not authorized to access this server.");
			return false;
		}

		if (record.status == RecordStatus.Found
				|| record.status == RecordStatus.Missing) {
			log.info("Officer "
					+ officerID
					+ " couldn't create a criminal record because the status was incompatible.");
			log.info(record);
			return false;
		}

		if (!assignCRecID(record)) {
			log.info("exceeded max id number");
			return false;
		}

		// to maximize concurrency, this line is not synchronized
		// records with different keys can access their corresponding list
		// concurrently
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));

		boolean success;

		synchronized (list) {
			success = list.add(record);
		}

		if (success) {
			log.info("Officer " + officerID
					+ " created a criminal record successfully.");
			log.info(record);
		} else {
			log.info("Officer " + officerID
					+ " couldn't created a criminal record.");
			log.info(record);
		}
		return success;
	}

	/**
	 * creates a missing record
	 * 
	 * @return true if record has been created successfully, false otherwise
	 */
	@Override
	public boolean createMRecord(String firstName, String lastName,
			String address, String lastDate, String lastLocation, int status,
			String officerID) {

		MissingRecord record = new MissingRecord(null, firstName, lastName,
				address, lastDate, lastLocation,
				EnumHelper.convertToRecordStatus(status));

		log.info("Officer " + officerID
				+ " requested to create a missing record.");
		log.info(record);

		if (!officerID.startsWith(name.toUpperCase())) {
			log.info("Officer " + officerID
					+ " is not authorized to access this server.");
			return false;
		}

		if (record.status == RecordStatus.OnTheRun
				|| record.status == RecordStatus.Captured) {
			log.info("Officer "
					+ officerID
					+ " couldn't create a missing record because the status was incompatible.");
			log.info(record);
			return false;
		}

		if (!assignMRecID(record)) {
			log.info("exceeded max id number");
			return false;
		}

		// to maximize concurrency, this line is not synchronized
		// records with different keys can access their corresponding list
		// concurrently
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));

		boolean success;

		synchronized (list) {
			success = list.add(record);
		}

		if (success) {
			log.info("Officer " + officerID
					+ " created a missing record successfully.");
			log.info(record);
		} else {
			log.info("Officer " + officerID
					+ " couldn't created a missing record.");
			log.info(record);
		}
		return success;
	}

	/**
	 * get records count from all servers
	 */
	@Override
	public String getRecordsCount(short recType, String officerID) {
		log.info("Officer " + officerID + " requested to get records count.");

		if (!officerID.startsWith(name.toUpperCase())) {
			log.info("Officer " + officerID
					+ " is not authorized to access this server.");
			return null;
		}

		// sends requests to other servers asking their recordsCount
		// wait to get response
		// add their count to your own records count and return
		int spvmCount = 0;
		int splCount = 0;
		int spbCount = 0;
		synchronized (this) {
			for (String remoteServer : serversInfo.keySet()) {
				if (remoteServer.equalsIgnoreCase("spvm")) {
					if (!remoteServer.equalsIgnoreCase(name)) {
						spvmCount = getRcordsCountFromPeer(serversInfo
								.get(remoteServer));
					} else {
						spvmCount = getRecordsCount();
					}
				} else if (remoteServer.equalsIgnoreCase("spl")) {
					if (!remoteServer.equalsIgnoreCase(name)) {
						splCount = getRcordsCountFromPeer(serversInfo
								.get(remoteServer));
					} else {
						splCount = getRecordsCount();
					}
				} else if (remoteServer.equalsIgnoreCase("spb")) {
					if (!remoteServer.equalsIgnoreCase(name)) {
						spbCount = getRcordsCountFromPeer(serversInfo
								.get(remoteServer));
					} else {
						spbCount = getRecordsCount();
					}
				}
			}
		}
		String count = "SPVM: " + spvmCount + ", SPL: " + splCount + ", SPB: "
				+ spbCount;
		log.info("Officer " + officerID + " received records count: " + count);
		return count;
	}

	/**
	 * get records count from other servers
	 * 
	 * @param port
	 * @return
	 */
	private int getRcordsCountFromPeer(int port) {
		DatagramSocket aSocket = null;
		try {
			aSocket = new DatagramSocket();
			byte[] m = REQUEST_RECORDS_COUNT.getBytes();
			InetAddress aHost = InetAddress.getByName("localhost");
			DatagramPacket request = new DatagramPacket(m,
					REQUEST_RECORDS_COUNT.length(), aHost, port);
			aSocket.send(request);
			byte[] buffer = new byte[UDP_BUFFER_SIZE];
			DatagramPacket reply = new DatagramPacket(buffer, buffer.length);
			aSocket.receive(reply);

			String msg = new String(reply.getData(), reply.getOffset(),
					reply.getLength());
			return Integer.parseInt(msg);
		} catch (SocketException e) {
			System.out.println("Error Socket: " + e.getMessage());
		} catch (IOException e) {
			System.out.println("Error IO: " + e.getMessage());
		} finally {
			if (aSocket != null)
				aSocket.close();
		}
		return -1;
	}

	/**
	 * edits a records
	 */
	@Override
	public boolean editCRecord(String lastName, String recordID, int newStatus,
			String officerID) {
		log.info("Officer " + officerID
				+ " requested to edit a criminal record.");
		log.info("lastName: " + lastName + " RecordID: " + recordID
				+ " newStatus: " + newStatus);

		if (!officerID.startsWith(name.toUpperCase())) {
			log.info("Officer " + officerID
					+ " is not authorized to access this server.");
			return false;
		}

		if (EnumHelper.convertToRecordStatus(newStatus) == RecordStatus.Found
				|| EnumHelper.convertToRecordStatus(newStatus) == RecordStatus.Missing) {
			log.info("Error in officer " + officerID
					+ "'s request. The new status \"" + newStatus
					+ "\" is not compatible with criminal record.");
			return false;
		}

		ArrayList<Record> list = database.get(Character.toUpperCase(lastName
				.charAt(0)));

		synchronized (list) {
			for (Record record : list) {
				if (record.id.equals(recordID)) {
					if (record instanceof CriminalRecord) {
						((CriminalRecord) record).status = EnumHelper
								.convertToRecordStatus(newStatus);
						log.info("Officer " + officerID
								+ "'s request to update the status for record "
								+ recordID + " succeeded.");
						return true;
					} else {
						log.info("Error in officer "
								+ officerID
								+ "'s request. The specified record is not a criminal record.");
						return false;
					}
				}
			}
			return false;
		}
	}

	/**
	 * transfers a record from this server to the remote server
	 */
	@Override
	public boolean transferRecord(String badgeID, String recordID,
			String remoteStationServerName) {

		log.info("transfering record with ID: " + recordID + " requested by "
				+ badgeID + " to " + remoteStationServerName);

		if (!badgeID.startsWith(name.toUpperCase())) {
			log.info("this officer does not have access to this server");
			return false;
		}

		Record record = getRecord(recordID);
		if (record == null) {
			log.info("record not found");
			return false;
		}
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));

		boolean success = false;

		synchronized (list) {
			int maxRequest = 10;
			while (!success && maxRequest > 0) {

				// send the object to the peer over UDP and wait for response
				success = sendObjectToPeer(record, remoteStationServerName);

				if (success) {
					log.info("transfering record was successful.");
					list.remove(record);
					log.info("record removed from database");
				} else {
					log.info("transfering record failed.");
					if (maxRequest > 1) {
						log.info("trying again ...");
					}
				}

				maxRequest--;
			}
		}
		return success;
	}

	/**
	 * receives a record from a remote server and redirects it to the
	 * appropriate method to do the transfer
	 * 
	 * @param record
	 *            record object for transfer
	 * @param name
	 *            receiver name
	 * @throws IOException
	 */
	protected boolean receiveTransfer(Record record, String name)
			throws IOException {
		if (record instanceof CriminalRecord) {
			return receiveTransferCR(record.id, record.firstName,
					record.lastName, ((CriminalRecord) record).description,
					record.status, name);
		} else if (record instanceof MissingRecord) {
			return receiveTransferMR(record.id, record.firstName,
					record.lastName, ((MissingRecord) record).address,
					((MissingRecord) record).lastDate,
					((MissingRecord) record).lastLocation, record.status, name);
		}
		return false;
	}

	/**
	 * receives a {@link CriminalRecord} from a remote server and puts it in its
	 * database
	 * 
	 * @return true if the record has been transferred successfully, false
	 *         otherwise
	 * @throws IOException
	 */
	protected boolean receiveTransferCR(String recordID, String firstName,
			String lastName, String description, RecordStatus status,
			String senderName) throws IOException {

		log.info("receiving a criminal record.");

		CriminalRecord record = new CriminalRecord(recordID, firstName,
				lastName, description, status);

		// to maximize concurrency, this line is not synchronized
		// records with different keys can access their corresponding list
		// concurrently
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));

		boolean success;

		synchronized (list) {
			success = list.add(record);
		}

		if (success) {
			log.info("record received and added to database successfully");
		} else {
			log.info("adding record failed.");
		}
		log.info(record.toString());
		return success;
	}

	/**
	 * receives a {@link MissingRecord} from a remote server and puts it in its
	 * database
	 * 
	 * @return true if the record has been transferred successfully, false
	 *         otherwise
	 * @throws IOException
	 */
	protected boolean receiveTransferMR(String recordID, String firstName,
			String lastName, String address, String lastDate,
			String lastLocation, RecordStatus status, String senderName)
			throws IOException {

		log.info("receiving a missing record ...");

		MissingRecord record = new MissingRecord(recordID, firstName, lastName,
				address, lastDate, lastLocation, status);

		// to maximize concurrency, this line is not synchronized
		// records with different keys can access their corresponding list
		// concurrently
		ArrayList<Record> list = database.get(Character
				.toUpperCase(record.lastName.charAt(0)));

		boolean success;

		synchronized (list) {
			success = list.add(record);
		}

		if (success) {
			log.info("record received and added to database successfully");
		} else {
			log.info("adding record failed.");
		}
		log.info(record.toString());
		return success;
	}

	/**
	 * send the record object to receiver over UDP
	 * 
	 * @param record
	 *            Record object
	 * @param name
	 *            receiver name
	 * @return true if the peer received the object correctly, false otherwise
	 */
	private synchronized boolean sendObjectToPeer(Record record, String name) {
		boolean result = false;
		int port = serversInfo.get(name);
		DatagramSocket socket = null;
		try {
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			ObjectOutputStream oos = new ObjectOutputStream(baos);

			oos.writeObject(record);
			oos.flush();
			// get the byte array of the object
			byte[] sendBuf = baos.toByteArray();

			socket = new DatagramSocket();
			DatagramPacket packet = new DatagramPacket(sendBuf, sendBuf.length,
					InetAddress.getByName("localhost"), port);
			// send the object
			socket.send(packet);
			log.info("record object sent.");

			byte[] recvBuf = new byte[UDP_BUFFER_SIZE];
			DatagramPacket reply = new DatagramPacket(recvBuf, recvBuf.length);
			// wait and receive reply
			socket.receive(reply);

			String msg = new String(reply.getData(), reply.getOffset(),
					reply.getLength());
			log.info("reply received: " + msg);

			String[] msgArray = msg.split(":");

			if (msgArray != null && msgArray.length == 2 && msgArray[1] != null
					&& msgArray[1].equals("1")) {
				result = true;
			} else {
				result = false;
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (socket != null) {
				socket.close();
			}
		}
		return result;
	}

	/**
	 * get record object with given id
	 * 
	 * @param id
	 * @return
	 */
	private Record getRecord(String id) {
		for (ArrayList<Record> list : database.values()) {
			for (Record record : list) {
				if (record.id.equals(id)) {
					return record;
				}
			}
		}
		return null;
	}

	class UDPServer implements Runnable {

		ConcurrentHashMap<Character, ArrayList<Record>> database;

		public int port;

		public UDPServer(int port) {
			this.port = port;
		}

		@Override
		public void run() {
			byte[] buffer = new byte[UDP_BUFFER_SIZE];
			DatagramSocket socket = null;
			try {
				socket = new DatagramSocket(port);
				while (true) {
					DatagramPacket request = new DatagramPacket(buffer,
							buffer.length);
					socket.receive(request);

					int len = request.getLength();
					String msg = new String(request.getData(),
							request.getOffset(), request.getLength());
					
					if (msg.equals(REQUEST_RECORDS_COUNT)) {
						byte[] replyBytes = ("" + getRecordsCount()).getBytes();
						DatagramPacket reply = new DatagramPacket(replyBytes,
								replyBytes.length, request.getAddress(),
								request.getPort());
						// send reply
						socket.send(reply);
					} else {

						Record record = getRecordObject(buffer);
						if (record != null) {
							// if the packet is a Record object
							boolean success = receiveTransfer(record,
									getServerNameByPort(request.getPort()));
							byte[] replyBytes = ("TransferResult:" + (success ? "1"
									: "0")).getBytes();
							DatagramPacket reply = new DatagramPacket(
									replyBytes, replyBytes.length,
									request.getAddress(), request.getPort());
							// send reply
							socket.send(reply);
						} else {
							log.info("wrong message received by " + name);
							log.info(msg);
							byte[] replyBytes = ("TransferResult:0").getBytes();
							DatagramPacket reply = new DatagramPacket(
									replyBytes, replyBytes.length,
									request.getAddress(), request.getPort());
							// send reply
							socket.send(reply);
						}
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				socket.close();
			}
		}

		/**
		 * 
		 * @return Record object if the received packet is a Record, null
		 *         otherwise
		 */
		private Record getRecordObject(byte[] buffer) {
			ByteArrayInputStream baos = new ByteArrayInputStream(buffer);
			ObjectInputStream oos;
			try {
				oos = new ObjectInputStream(baos);
				Record record = (Record) oos.readObject();
				return record;
			} catch (Exception e) {
				e.printStackTrace();
			}
			return null;
		}

	}

}
